package com.eversec.msg.util;

import com.eversec.kafka.consumer.KafkaConsumerServer;
import com.eversec.kafka.exception.KafkaInitException;
import com.eversec.kafka.producer.KafkaProducerServer;
import com.eversec.msg.util.props.PropertiesInfo;
import org.apache.log4j.Logger;

import java.util.List;

/**
 * Created by eversec on 2018/2/26.
 */
public class KafkaUtil {

    private static Logger logger=Logger.getLogger(KafkaUtil.class);

    private static KafkaProducerServer<List<String>> kafkaProducerServer=null;

    /**
     * 单例模式：如果每次发生数据都创建一个tcp 这样连接数会很多
     * @return
     * @throws KafkaInitException
     */
    public static KafkaProducerServer<List<String>> getKafkaProducerServer() throws KafkaInitException{
        if(kafkaProducerServer==null){
            kafkaProducerServer=new KafkaProducerServer<List<String>>(PropertiesInfo.KAFKA_PRODUCER_HOST);
        }
        return kafkaProducerServer;
    }

    /**
     * 消费者
     * @param topic
     */
    public static KafkaConsumerServer getKafkaConsumerServerByTopic(String topic) {
        KafkaConsumerServer kafkaConsumerServer = null;
        try {
            kafkaConsumerServer = new KafkaConsumerServer(PropertiesInfo.KAFKA_CONSUMER_HOST,topic);
        } catch (KafkaInitException e) {
            logger.error("创建kafka消费者失败,获取kafka消息异常返回", e);
        }
        logger.info("创建消费者");
//		kafkaConsumerServer.start();
        return kafkaConsumerServer;
    }

    /**
     * 发送数据
     * @param list
     * @param list
     */
    public  static void setToTZDBtype(List<String> list,String topic) {
        try {
            getKafkaProducerServer().sendMessages(topic, null, list);
            logger.info("kafka 生产更新自动下发升级包定时任务时间消息：【topic ="+topic+"--size ="+list.size()+"】");
        } catch (KafkaInitException e) {
            // TODO Auto-generated catch block
            e.printStackTrace();
        }
    }

}

